package com.atguigu.chapter08;

import com.atguigu.Bean.HotItem;
import com.atguigu.Bean.UserBehavior;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.*;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Date;
import java.util.TreeSet;

/**
 * @ClassName: Flink01_PV_Project
 * @Description:
 * @Author: kele
 * @Date: 2021/4/9 18:52
 *
 *
 * 需求：
 *每隔10分钟输出最近1小时内点击量最多的前N个商品
 * 	最近一小时: 窗口长度
 * 	每隔10分钟: 窗口滑动步长
 * 	时间: 使用event-time
 *
 * 1、获取数据，过滤出pv类型的，封装为UserBehavior类型
 * 2、根据商品进行keyby
 * 3、开滑动窗（窗口1h，步长10min）   --开窗只跟key相关
 * 4、统计每个商品的点击量
 *                --注意：每个商品一个窗口，无法对同一时间的商品进行聚合，
 *                --所以之后要将在同一时间段的商品再进行keyby
 *                --统计的时候需要agg函数，返回值中带窗口的结束时间
 * 5、再按照每个商品所在窗口的end时间进行keyby
 * 6、统计商品数量在前N的商品
 *
 *
 **/
public class Flink03_TopN_Project {

    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port",20000);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        env.setParallelism(2);

//        1、获取数据，过滤出pv类型的，封装为UserBehavior类型
        env.readTextFile("in/UserBehavior.csv")
                .flatMap(new FlatMapFunction<String, UserBehavior>() {

                    @Override
                    public void flatMap(String line, Collector<UserBehavior> out) throws Exception {

                        String[] split = line.split(",");
                        UserBehavior behavior = new UserBehavior(Long.valueOf(split[0]),
                                                                Long.valueOf(split[1]), Integer.valueOf(split[2]),
                                                                split[3], Long.valueOf(split[4]));

                        if("pv".equals(behavior.getBehavior())){
                            out.collect(behavior);
                        }

                    }
                })
                //添加水印
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<UserBehavior>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner(new SerializableTimestampAssigner<UserBehavior>() {
                            @Override
                            public long extractTimestamp(UserBehavior element, long recordTimestamp) {
                                return element.getTimestamp() * 1000L;
                            }
                        })
                )
//                2、按照商品进行keyby，统计在一小时的窗口中商品的数目
                .keyBy(UserBehavior::getItemId)
                .window(SlidingEventTimeWindows.of(Time.hours(2),Time.minutes(30)))
                .aggregate(new AggregateFunction<UserBehavior, Long, Long>() {
                               @Override
                               public Long createAccumulator() {
                                   return 0l;  //初始化
                               }

                               @Override
                               public Long add(UserBehavior value, Long acc) {
                                   return acc + 1l; //来一条数据加一
                               }

                               @Override
                               public Long getResult(Long acc) {
                                   return acc;
                               }

                               @Override
                               public Long merge(Long a, Long b) {
                                   return a + b; //在session窗口中使用
                               }

                           },

//                        3、由于后续需要将同一窗口的数据放一起统计keyby，所以输出时需要包含窗口的时间，
                        //这里的long的key是之前keyby的key
                        new ProcessWindowFunction<Long, HotItem, Long, TimeWindow>() {
                            @Override
                            public void process(Long key,
                                                Context ctx,
                                                Iterable<Long> elements,
                                                Collector<HotItem> out) throws Exception {

                                out.collect(new HotItem(key,
                                                        elements.iterator().next(),
                                                        ctx.window().getEnd()
                                        ));

                            }
                        }
                )
//                4、按照同一个窗口keyby，计算同一窗口下点击数量最多的3个商品

                .keyBy(HotItem::getWindowEndTime)
                .process(new KeyedProcessFunction<Long, HotItem, String>() {
                    private ValueState<Long> isFirst;
                    private ListState<HotItem> hotItemState;

                    //将结果保存成一个list状态


                    @Override
                    public void open(Configuration parameters) throws Exception {

                        //保存数据的状态
                        hotItemState =
                                getRuntimeContext().getListState(new ListStateDescriptor<HotItem>("HotItemState", HotItem.class));

                     //判断第一条数据到达的状态，也可以使用hotItemState的状态，判断hotItemState大小为0，
                     // 但hotItemState需要遍历，再转化为list才可以统计数目
                     // 所以这里单独定义一个状态
                        isFirst = getRuntimeContext().getState(new ValueStateDescriptor<Long>("isFirst", Long.class));

                    }

                    @Override
                    public void processElement(HotItem value,
                                               Context ctx,
                                               Collector<String> out) throws Exception {

                        //当第一条数据来的时候，启动定时器，当时间到1h时最后一条数据到达
                        //判断在isFirst中的值为null，则说明是第一次
                        if(isFirst.value() == null){

                            long timeFlag = value.getWindowEndTime() + 1000;

                            ctx.timerService().registerEventTimeTimer(timeFlag);

                            isFirst.update(timeFlag);
                        }

                        hotItemState.add(value);

                    }

//                    定时器时间到，对所有的数据进行排序处理
                     @Override
                     public void onTimer(long timestamp,
                                         OnTimerContext ctx,
                                         Collector<String> out) throws Exception {

                         TreeSet<HotItem> set = new TreeSet<>(new Comparator<HotItem>() {
                             @Override
                             public int compare(HotItem o1, HotItem o2) {
                                 int result = o2.getCount().compareTo(o1.getCount());
                                 return  result == 0 ? 1 : result;

                             }
                         });

                         for (HotItem hotItem : hotItemState.get()) {
                             set.add(hotItem);

                             if(set.size() > 3){
                                 set.pollLast();
                             }
                         }

                         StringBuilder s = new StringBuilder();
                         s.append("-----------------\n");
                         s.append("窗口结束时间: ").append(timestamp - 10000L).append("\n");
                         for (Object o : set) {
                             s.append(o).append("\n");
                         }

                         out.collect(s.toString());

                         hotItemState.clear();
                         isFirst.clear();

                     }
                })
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
